package com.rpcservice.rpc.remoting.transport.netty.client;

import com.rpcservice.rpc.common.enums.CompressTypeEnum;
import com.rpcservice.rpc.common.enums.SerializationTypeEnum;
import com.rpcservice.rpc.common.factory.SingletonFactory;
import com.rpcservice.rpc.remoting.constants.RpcConstants;
import com.rpcservice.rpc.remoting.dto.RpcMessage;
import com.rpcservice.rpc.remoting.dto.RpcMessageType;
import com.rpcservice.rpc.remoting.dto.RpcResponse;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.ReferenceCounted;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

/**
 * 自定义客户端 ChannelHandler 以处理服务器发送的数据
 * <p>
 * 如果继承自 SimpleChannelInboundHandler 的话就不要考虑 ByteBuf 的释放 ，{@link SimpleChannelInboundHandler} 内部的
 * * channelRead 方法会替你释放 ByteBuf ，避免可能导致的内存泄露问题。
 * <p>
 * 继承自 ChannelInboundHandlerAdapter，
 * 这是 Netty 提供的一个简单的 ChannelInboundHandler 实现，
 * 它可以帮助我们更方便地处理入站事件，
 * 比如读取数据、处理数据、发送响应等。
 * 我们只需要实现 channelRead0 方法即可，
 * 这个方法会在 Channel 中有数据可读时被调用。
 */
public class NettyRpcClientHandler extends ChannelInboundHandlerAdapter {
    private static final Logger logger = LoggerFactory.getLogger(NettyRpcClientHandler.class);
    private final UnprocessedRequests unprocessedRequests;
    private final NettyRpcClient nettyRpcClient;

    public NettyRpcClientHandler() {
        this.nettyRpcClient = SingletonFactory.getInstance(NettyRpcClient.class);
        this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);
    }

    /**
     * 这是 Netty 提供的一个 ChannelInboundHandler 接口的方法，
     * 它会在 Channel 中有数据可读时被调用。
     * 我们可以在这个方法中处理数据。。
     * 然后通过 ctx.writeAndFlush 方法将结果写入到输出缓冲区并发送给客户端。
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            logger.info("client receive msg: [{}]", msg);
            if (msg instanceof RpcMessage) {
                RpcMessage tmpMsg = (RpcMessage) msg;
                byte messageType = tmpMsg.getMessageType();
                switch (messageType) {
                    case RpcConstants.HEARTBEAT_REQUEST_TYPE -> logger.info("heart [{}]", tmpMsg.getData());
                    case RpcConstants.REQUEST_TYPE -> {
                        RpcResponse<Object> response = (RpcResponse<Object>) tmpMsg.getData();
                        unprocessedRequests.complete(response);
                    }
                }
            }
        } finally {
            //确保ByteBuf被释放，避免内存泄漏
            //如果msg是ReferenceCounted类型的，说明它是一个引用计数对象，
            //可以通过调用release()方法来释放它的内存。
            //如果msg不是ReferenceCounted类型的，说明它不是一个引用计数对象，
            //那么就不需要调用release()方法来释放它的内存。
            ReferenceCountUtil.release(msg);
        }
    }

    /**
     * Netty 心跳机制相关。保证客户端和服务端的连接不被断掉，避免重连，
     *
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.WRITER_IDLE) {
                logger.info("write idle happen [{}]", ctx.channel().remoteAddress());
                Channel channel = nettyRpcClient.getChannel((InetSocketAddress) ctx.channel().remoteAddress());
                RpcMessage rpcMessage = new RpcMessage();
                rpcMessage.setCodec(SerializationTypeEnum.PROTOSTUFF.getCode());
                rpcMessage.setCompress(CompressTypeEnum.GZIP.getCode());
                rpcMessage.setMessageType(RpcConstants.HEARTBEAT_REQUEST_TYPE);
                rpcMessage.setData(RpcConstants.PING);
                channel.writeAndFlush(rpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }
    }

    /**
     * Called when an exception occurs in processing a client message
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("client caught exception", cause);
        cause.printStackTrace();
        ctx.close();
    }
}
